package org.gorson.executor;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;

import java.util.Arrays;

public class demo1 {

    public static void main(String[] args) {
        EnvironmentSettings settings = EnvironmentSettings.inStreamingMode();
        TableEnvironment env = TableEnvironment.create(settings);

        env.executeSql("CREATE TABLE transactions (\n" +
                "    account_id  BIGINT,\n" +
                "    amount      BIGINT,\n" +
                "    transaction_time TIMESTAMP(3),\n" +
                "    WATERMARK FOR transaction_time AS transaction_time - INTERVAL '5' SECOND\n" +
                ") WITH (\n" +
                "    'connector' = 'kafka',\n" +
                "    'topic'     = 'transactions',\n" +
                "    'properties.bootstrap.servers' = '172.16.44.99:9092',\n" +
                "    'format'    = 'csv'\n" +
                ")");

        env.executeSql("CREATE TABLE spend_report (\n" +
                "    account_id BIGINT,\n" +
                "    log_ts     TIMESTAMP(3),\n" +
                "    amount     BIGINT\n," +
                "    PRIMARY KEY (account_id, log_ts) NOT ENFORCED" +
                ") WITH (\n" +
                "   'connector'  = 'jdbc',\n" +
                "   'url'        = 'jdbc:mysql://localhost:3306/flink',\n" +
                "   'table-name' = 'spend_report',\n" +
                "   'driver'     = 'com.mysql.jdbc.Driver',\n" +
                "   'username'   = 'root',\n" +
                "   'password'   = '123456'\n" +
                ")");

        Table transactions = env.from("spend_report");
//        report(transactions).executeInsert("spend_report");
//        transactions.executeInsert("spend_report");
        env.executeSql("select * from spend_report").print();

        String[] catlogs = env.listCatalogs();
        String[] db = env.listDatabases();
        String[] table = env.listTables();
        System.out.println(Arrays.toString(catlogs));
        System.out.println(Arrays.toString(db));
        System.out.println(Arrays.toString(table));

    }

}
